package com.intct.dim;

import com.alibaba.fastjson.JSONObject;
import com.intct.common.Constant;
import com.intct.func.HBaseSinkFunction;
import com.intct.util.FromPropertiesFileUtil;
import com.intct.util.HBaseUtil;
import com.intct.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hbase.thirdparty.com.google.common.reflect.Parameter;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * @author gufg
 * @since 2025-07-27 11:14
 * 抽取维度表数据
 */
public class DimApp {
    
    private static ParameterTool parameterTool = FromPropertiesFileUtil.parameterTool;
    
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 配置源
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(Constant.KAFKA_ODS_DB_TOPIC, "dim-app-id");
        DataStreamSource<String> sourceDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "dim-job-app");

        
        
        // 从Kafka获取的数据,怎么判断是维度/码表/业务事实表  ????
        // area_info driver_info order_state ...
        List<String> tableList = Arrays.asList(parameterTool.get("dim-tables").split(","));

        // 过滤掉非维度表
        SingleOutputStreamOperator<String> filterDS = sourceDS.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String jsonStr) throws Exception {
                // 转为JSONObject
                JSONObject jsonObj = JSONObject.parseObject(jsonStr);
                String table = jsonObj.getString(Constant.MYSQL_CDC_JSON_KEY_SOURCE_TABLE);
                if (!tableList.contains(table)) {
                    return false;
                }
                return true;
            }
        });

        // 维度表: id,name,age,sex ==> id,name???
        List<String> orderStateCol = Arrays.asList(parameterTool.get("dim-order-state-table-col").split(","));

        // {"op":"r","after":{"name":"已下单","id":1,"state":"1001"},"ts_ms":1753585266163,"db":"travel","table":"order_state"}
        // sink hbase: 表名(dim_order_state) 列族(info) 列名 值

        SingleOutputStreamOperator<JSONObject> mapDS = filterDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String jsonStr) throws Exception {
                // 转换成JSONObject
                JSONObject jsonObj = JSONObject.parseObject(jsonStr);

                // 操作类型(CRUD)
                String op = jsonObj.getString(Constant.MYSQL_CDC_JSON_KEY_OP);

                JSONObject currentJSONObj = null;

                // 如果是删除数据,则从before获取数据
                if (op.equals("d")) {
                    // 获取当前值
                    currentJSONObj = jsonObj.getJSONObject(Constant.MYSQL_CDC_JSON_KEY_BEFORE);
                    jsonObj.remove(Constant.MYSQL_CDC_JSON_KEY_BEFORE);
                } else {
                    // 获取当前值
                    currentJSONObj = jsonObj.getJSONObject(Constant.MYSQL_CDC_JSON_KEY_AFTER);
                    jsonObj.remove(Constant.MYSQL_CDC_JSON_KEY_AFTER);
                }

                // 字段对比
                // 处理之前: "after":{"name":"已下单","id":1,"state":"1001"}
                // 处理之后: "after":{"name":"已下单","state":"1001"}
                // 处理之后(整体): {"op":"r","data":{"name":"已下单","state":"1001"},"ts_ms":1753585266163,"db":"travel","table":"order_state"}
                Set<Map.Entry<String, Object>> entries = currentJSONObj.entrySet();
                entries.removeIf(entrie -> !orderStateCol.contains(entrie.getKey()));

//                for (Map.Entry<String, Object> entry : entries) {
//                    String jsonKey = entry.getKey();
//                    if (!orderStateCol.contains(jsonKey)) {
//                        entries.remove(jsonKey);
//                    }
//                }
                jsonObj.put("data", currentJSONObj);

                return jsonObj;
            }
        });

        // sink hbase
        // create hbase table
        mapDS.process(new ProcessFunction<JSONObject, JSONObject>() {
            @Override
            public void processElement(JSONObject jsonObj, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                // {"op":"r","data":{"name":"已结束","state":"3002"},"ts_ms":1753585266172,"db":"travel","table":"order_state"}

                // 创建HBase表
                String namespace = "intct82";
                String tableName = "dim_order_state";
                // 列族名称
                String familyName = "info";

                // 执行创建HBase表
                HBaseUtil.createTable(namespace, tableName, familyName);

                out.collect(jsonObj);
            }
        });

        // insert into
        mapDS.addSink(new HBaseSinkFunction("intct82", "dim_order_state", "info", "state"));

        env.execute();
    }
}
